package com.ibm.cps.servlet;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.kafka.KafkaProducer;
import com.ibm.cps.kafka.KafkaTopicConstructorForMultiTenants;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.validation.ValidatorUtil;
import com.ibm.cps.model.*;
import com.ibm.cps.newmessage.AbstractDataSourceMetadata;
import com.ibm.cps.newmessage.DataSourceMetadataFactory;
import com.ibm.factories.PublishProcessorFactory;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;

/**
 * Created by gongxuan on 12/2/15.
 */
public class DataSourceImplementationV2 implements ServletTemplate {
    private Integer kafkaPartition;
    private Logger logger;
    private KafkaProducer producer;
    private IPersistency persistency;
    private DataSourceMonitor monitor;
    private KafkaTopicConstructorForMultiTenants kafkaTopicConstructorForMultiTenants = new KafkaTopicConstructorForMultiTenants();
    public DataSourceImplementationV2() throws CPSException {
        kafkaPartition = LocalConfig.KAFKA_PARTITION;
//		monitor = DataSourceMonitor.getInstance();
        monitor = DataSourceMonitor.getInstance();
        logger = LoggerFactory.getLogger(this.getClass());
        producer = new KafkaProducer(LocalConfig.KAFKA_BROKERS);
        persistency = MongoPersistency.getInstace();
        logger.info("Connect to kafka: " + LocalConfig.KAFKA_BROKERS);
        logger.info("Connect to mongo: " + LocalConfig.MONGO_CONNECT_STRING + ", and database is "
                + LocalConfig.PERSISTENCY_DATABASE);
    }

    @Override
    public String doPost(Map<String, String> key2Value, String input) throws CPSException {
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
        String source = key2Value.get("source");
        if (source == null) {
            source = "rest";
        }
        return validateDataSource(tenantid, input);
    }

    @Override
    public String doGet(Map<String, String> key2Value) throws CPSException {
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
        String source = key2Value.get(MessageFields.SOURCE);
        if (source == null) {
            source = "rest";
        }

        PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
        MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
        Collection<String> metadatas = persistency.getTenantDataSource(tenantid, key2Value.get(MessageFields.NAME));
        Collection<String> result = adapter.getJsons(metadatas, metadatas);
        return ServletUtil.constructCollectionJson(result);
    }

    @Override
    public String doDelete(Map<String, String> key2Value, String input) throws CPSException {
        throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "This method is not supported now.");
    }

    @Override
    public String doDelete(Map<String, String> key2Value) throws CPSException {
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
        String source = key2Value.get(MessageFields.SOURCE);
        if (source == null) {
            source = "rest";
        }
        Collection<String> metadatas = persistency.getDataSource(tenantid, key2Value.get(MessageFields.NAME));
        if (metadatas != null) {
            for (String dataSource : metadatas) {
                AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
                String name = metadata.getName();
                ObjectNode node = JSONUtility.newObjectNode();
                node.set(MessageFields.PARENTPROCESSORS, JSONUtility.newArrayNode().add(name));

                Collection<String> inputs = persistency.getProcessors(node);
                if (inputs != null && inputs.size() >= 1) {
                    throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Can't delete this data source "
                            + metadata.getName() + " because it's referenced by other processor metadatas.");
                }
            }
            List<String> deleted = new ArrayList<String>();

            try {
                for (String dataSource : metadatas) {
                    AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
                    monitor.deleteDataSource(tenantid, metadata);
                    Collection<String> result = persistency.deleteDataSource(tenantid, metadata.getName());
                    //broadcast
                    for(int i= 0;i<kafkaPartition;i++)
                    {
                        producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceDeleteTopic(tenantid),
                                i,metadata.getInput().getBytes());
                    }
                    if (result != null) {
                        for (String ds : result) {
                            deleted.add(ds);
                        }
                    }
                }
            } catch (Throwable e) {

                // transactionality
                for (String dataSource : deleted) {
                    monitor.validateDataSource(tenantid, JSONUtility.fromObjectString(dataSource));
                    persistency.putDataSource(dataSource);
                    //broadcast
                    for(int i= 0;i<kafkaPartition;i++)
                    {
                        producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(tenantid),
                                i,dataSource.getBytes());
                    }
                }
                throw e;
            }
            return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, ServletUtil.constructCollectionJson(deleted));
        }
        return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted");
    }

    @Override
    public String doPut(Map<String, String> key2Value, String input) throws CPSException {
        throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Put method for data source is unsupported.");
    }

    /**
     * validate the data source metadata and put it in persistency and send to
     * kafka
     *
     * @param tenantid
     * @param input
     * @return
     * @throws CPSException
     */
    public String validateDataSource(String tenantid, String input) throws CPSException {
        if (input == null || input.length() < 1) {
            throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Nothing to be posted.");
        }
        ArrayList<AbstractDataSourceMetadata> sendObjects = new ArrayList<AbstractDataSourceMetadata>();
        try {
            ArrayNode array = JSONUtility.getArrayFromString(input);
            ArrayNode retArray = JSONUtility.newArrayNode();

            for (int i = 0; i < array.size(); i++) {
                ObjectNode objectNode = (ObjectNode) array.get(i);
                AbstractDataSourceMetadata dataSource = monitor.validateDataSource(tenantid, objectNode);
                ObjectNode retNode = JSONUtility.newObjectNode();
                retNode.put("id", dataSource.getName());
                retArray.add(retNode);
                sendObjects.add(dataSource);
            }

            for (int i = 0; i < sendObjects.size(); i++) {
                persistency.putDataSource(sendObjects.get(i).getInput());
                //broadcast
                for(int j= 0;j<kafkaPartition;j++)
                {
                    producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(tenantid),
                            j,sendObjects.get(i).getInput().getBytes());
                }
            }
            return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, retArray);
        } catch (CPSException e) {
            for (int i = 0; i < sendObjects.size(); i++) {
                AbstractDataSourceMetadata tmpObject = sendObjects.get(i);
                persistency.deleteDataSource(tenantid, tmpObject.getName());
                monitor.deleteDataSource(tenantid, tmpObject);
            }
            throw e;
        }
    }
}

